package cn.cnic.security.nssap.stearm

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.io.InputStream
import java.util.Properties

object CommonTools {

  /**
    * 设置flink的kafka消费者
    *
    * @param topic 主题
    * @param env   StreamExecutionEnvironment
    * @return DataStream
    */
  def kafkaConsumer(topic: String, env: StreamExecutionEnvironment): DataStream[String] = {
    //从配置文件读取kafka设置
    val properties = new Properties()
    val kafkaInputStream: InputStream = getClass.getClassLoader.getResourceAsStream("kafka.properties")
    properties.load(kafkaInputStream)

    val schema = new SimpleStringSchema()
    //获得FlinkKafka连接器
    val flinkKafkaConsumer = new FlinkKafkaConsumer[String](topic, schema, properties)

    env.addSource(flinkKafkaConsumer)

  }

}
